package com.mqtt.config;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

/**
 * @Author 咕唧
 * @Date 2022/7/30 15:56
 * @Version 1.0
 */
@Configuration
public class MyConfig {


    /**
     * 连接工厂
     *
     * @return
     */
    @Bean
    public MqttPahoClientFactory mqttPahoClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();

        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{"tcp://120.48.95.238:1883"});
        options.setCleanSession(false);

        //关闭主机名认证
        options.setHttpsHostnameVerificationEnabled(false);
        options.setUserName("dc3");
        options.setPassword("dc3".toCharArray());


        options.setKeepAliveInterval(15);
        factory.setConnectionOptions(options);
        return factory;
    }

    //mqttOutputChannel 代表的是发消息的通道对象的id
    @Bean
    public MessageChannel mqttOutputChannel() {
        return new DirectChannel();
    }

    /**
     * 发送消息的通道对象 mqttOutputChannel表示发送消息对象的id
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutputChannel")
    public MessageHandler mqttOutbound(MqttPahoClientFactory mqttPahoClientFactory) {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("testId", mqttPahoClientFactory);
        messageHandler.setAsync(true);
        //默认的qos
        messageHandler.setDefaultQos(1);
        messageHandler.setDefaultTopic("yigeduilie");
        return messageHandler;
    }


    /**
     * 收消息的通道,注意实际开发中和发的可能不在一起
     *
     * @return
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }


    @Bean
    public MessageProducer inbound(MessageChannel mqttInputChannel, MqttPahoClientFactory mqttPahoClientOutFactory) {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter("xiaofeizhe", mqttPahoClientOutFactory,
                        "yigeduilie");
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel);
        return adapter;
    }

    /**
     * 收消息的处理器,用于如何处理消息
     * mqttInputChannel代表收消息的通道的id
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.err.println(message.getHeaders());
                System.err.println(message.getPayload());
            }

        };
    }

}
